package com.iot.app.kafka.util;

import org.apache.kafka.common.serialization.Serializer;
import org.apache.log4j.Logger;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.iot.app.kafka.vo.IoTData;

import java.util.Map;

/**
 * Class to convert IoTData java object to JSON String
 * 
 * @author ryo yang
 *
 */
/*public class IoTDataEncoder implements Encoder<IoTData> {

	private static final Logger logger = Logger.getLogger(IoTDataEncoder.class);
	private static ObjectMapper objectMapper = new ObjectMapper();
	public IoTDataEncoder(VerifiableProperties verifiableProperties) {

    }

	@Override
	public byte[] toBytes(IoTData iotEvent) {
		try {
			String msg = objectMapper.writeValueAsString(iotEvent);
			logger.info(msg);
			return msg.getBytes();
		} catch (JsonProcessingException e) {
			logger.error("Error in Serialization", e);
		}
		return null;
	}
}*/

public class IoTDataEncoder implements Serializer<IoTData> {

	private static final Logger logger = Logger.getLogger(IoTDataEncoder.class);
	private static ObjectMapper objectMapper = new ObjectMapper();

	@Override
	public void configure(Map configs, boolean isKey) {

	}

	@Override
	public byte[] serialize(String s, IoTData iotEvent) {
		try {
			if (iotEvent == null) {
				return null;
			}
			else {
				String msg = objectMapper.writeValueAsString(iotEvent);
				logger.info(msg);
				return msg.getBytes();
			}
		} catch (JsonProcessingException e) {
			logger.error("Error in Serialization", e);
		}

		return null;
	}

	@Override
	public void close() {

	}
}


